package com.example.rocktmqsample;

import com.alibaba.fastjson.JSON;
import com.example.rocktmqsample.produce.MyTransactionListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@Slf4j
@RestController
@SpringBootApplication
public class RocketmqSampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(RocketmqSampleApplication.class, args);
    }

    @Autowired
    private RocketMQTemplate rocketmqTemplate;

    @GetMapping("/sync")
    public String sync() {
        SendResult sendResult = rocketmqTemplate.syncSend("topic_1", "同步消息");
        if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
            return "同步发送成功";
        }
        return "同步发送失败";
    }

    @RequestMapping("/orderly")
    public String orderly() {
        rocketmqTemplate.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object o) {
                return list.get(1);
            }
        });
        rocketmqTemplate.syncSendOrderly("orderly", "顺序消息", "123");
        return "发送顺序消息";
    }

    @RequestMapping("/delay")
    public String delay() {
        rocketmqTemplate.asyncSend("delay_topic", MessageBuilder.withPayload("Hello, RocketMQ!").build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 发送成功的回调处理
                log.info("延时消息发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                // 发送异常的回调处理
                log.info("延时消息发送异常");
            }
        }, 10);
        return "发送延时消息";
    }

    @GetMapping("transaction")
    public String transaction(){
        // 注册事务监听器
        Message<String> message = MessageBuilder.withPayload("事务消息")
                .setHeader(RocketMQHeaders.KEYS, 1)
                .setHeader("money", 10)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, 100)
                .build();
        TransactionSendResult transactionSendResult = rocketmqTemplate.sendMessageInTransaction("transaction_topic", message, null);
        LocalTransactionState localTransactionState = transactionSendResult.getLocalTransactionState();
        log.info("事务消息返回结果：{}", JSON.toJSONString(localTransactionState));
        return "事务消息";
    }

    @GetMapping("/retry")
    public String retry() {
        rocketmqTemplate.convertAndSend("retry_topic","消息重试");
        return "消息重试";
    }

}
